package org.zhaya.sdk.processors;

import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import io.logz.sawmill.Doc;
import io.logz.sawmill.ProcessResult;
import io.logz.sawmill.Processor;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.exceptions.ProcessorConfigurationException;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.zhaya.sdk.DodpProcessor;
import org.zhaya.sdk.util.JsonProcessorsUtils;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @Author Ryan Yang
 * @Date 2022/10/17 6:58 下午
 */
@ProcessorProvider(type = "columnsToJSONString", factory = ColumnsToJSONStringProcessor.Factory.class)
public class ColumnsToJSONStringProcessor implements Processor {

    private final String targetField;
    private final List<String> sourceFields;

    public ColumnsToJSONStringProcessor(String targetField, List<String> sourceFields) {
        this.targetField = targetField;
        this.sourceFields = sourceFields;
    }

    @Override
    public ProcessResult process(Doc doc) throws InterruptedException {
        String value = null;
        if (CollectionUtils.isNotEmpty(sourceFields)) {
            Map<String, Object> map = ColumnsToJSONStringProcessor.getMapFromSourceFields(doc, sourceFields);
            if (MapUtils.isNotEmpty(map)) {
                value = JsonProcessorsUtils.toJsonString(map);
            }
        } else {
            value = JsonProcessorsUtils.toJsonString(doc.getSource());
        }

        doc.removeField(targetField);
        if (Objects.nonNull(value)) {
            doc.addField(targetField, value);
        }
        return ProcessResult.success();
    }

    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Configuration implements DodpProcessor.Configuration {
        private String targetField;
        private List<String> sourceFields;

    }

    @NoArgsConstructor
    public static class Factory implements DodpProcessor.Factory {

        @Override
        public Processor create(Map<String, Object> config) {
            Configuration columnsToJSONStringConfig = JsonProcessorsUtils.fromJsonMap(Configuration.class, config);
            if (StrUtil.isBlank(columnsToJSONStringConfig.targetField)) {
                throw new ProcessorConfigurationException("The targetField field can not be blank.");
            }
            return new ColumnsToJSONStringProcessor(columnsToJSONStringConfig.targetField, columnsToJSONStringConfig.sourceFields);
        }
    }

    private static Map<String, Object> getMapFromSourceFields(Doc doc, List<String> sourceFields) {
        Map<String, Object> map = null;
        if (Objects.nonNull(doc) && CollectionUtils.isNotEmpty(sourceFields)) {
            map = Maps.newHashMapWithExpectedSize(sourceFields.size());
            for (String field : sourceFields) {
                if (doc.hasField(field)) {
                    map.put(field, doc.getField(field));
                }
            }
        }
        return map;
    }
}
